package com.zidongxiangxi.sharding.job.core;

import com.zidongxiangxi.sharding.job.core.exception.RegException;
import com.zidongxiangxi.sharding.job.core.exception.RegExceptionHandler;
import com.zidongxiangxi.sharding.job.core.executor.ShardingJobExecutor;
import com.zidongxiangxi.sharding.job.core.executor.ThreadPoolShardingJobExecutor;
import com.zidongxiangxi.sharding.job.core.executor.WaitingExecuteShardingItem;
import com.zidongxiangxi.sharding.job.core.executor.runnable.ShardingJobRunnable;
import com.zidongxiangxi.sharding.job.core.listener.JobTreeCacheListener;
import com.zidongxiangxi.sharding.job.core.listener.RegistryCenterConnectionStateListener;
import com.zidongxiangxi.sharding.job.core.node.AbstractNode;
import com.zidongxiangxi.sharding.job.core.node.ConfigNode;
import com.zidongxiangxi.sharding.job.core.node.InstancesNode;
import com.zidongxiangxi.sharding.job.core.node.LeaderNode;
import com.zidongxiangxi.sharding.job.core.node.ShardingNode;
import com.zidongxiangxi.sharding.job.core.listener.ConfigNodeChangeListener;
import com.zidongxiangxi.sharding.job.core.listener.InstancesNodeChangeListener;
import com.zidongxiangxi.sharding.job.core.listener.LeaderNodeChangeListener;
import com.zidongxiangxi.sharding.job.core.listener.ShardingNodeChangeListener;
import com.zidongxiangxi.sharding.job.core.registry.ZookeeperRegistryCenter;
import com.zidongxiangxi.sharding.job.core.storage.NodeStorage;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.api.transaction.CuratorTransactionFinal;
import org.apache.curator.framework.recipes.cache.TreeCache;

import java.util.Arrays;
import java.util.Collections;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Objects;

/**
 * 分片任务
 *
 * @author chenxudong
 * @date 2021/02/18
 */
@Slf4j
public class ShardingJob {
    @Setter
    private ShardingJobConfig config;

    private final ZookeeperRegistryCenter regCenter;
    private final ShardingJobExecutor shardingJobExecutor;

    @Getter
    private final String jobName;
    @Getter
    private final ConfigNode configNode;
    @Getter
    private final LeaderNode leaderNode;
    @Getter
    private final InstancesNode instancesNode;
    @Getter
    private final ShardingNode shardingNode;

    private volatile boolean started = false;
    private volatile boolean paused = false;

    public ShardingJob(
        final ShardingJobConfig config,
        final ZookeeperRegistryCenter regCenter,
        final ShardingJobRunnable runnable
    ) {
       this(config, regCenter, new ThreadPoolShardingJobExecutor(config.getJobName(), runnable));
    }

    public ShardingJob(
        final ShardingJobConfig config,
        final ZookeeperRegistryCenter regCenter,
        final ThreadPoolShardingJobExecutor shardingJobExecutor
    ) {
        this.jobName = config.getJobName();
        this.config = config;
        this.regCenter = regCenter;
        this.shardingJobExecutor = shardingJobExecutor;

        NodeStorage storage = new NodeStorage(jobName, regCenter);
        this.configNode = new ConfigNode(jobName, storage);
        this.leaderNode = new LeaderNode(jobName, storage);
        this.instancesNode = new InstancesNode(jobName, storage);
        this.shardingNode = new ShardingNode(jobName, storage);
    }

    public synchronized void start() {
        ShardingJobRegistry.getInstance().registerRegistryCenter(config.getJobName(), regCenter);
        configNode.setupJobConfig(config);

        ConfigNodeChangeListener configNodeChangeListener = new ConfigNodeChangeListener(this);
        InstancesNodeChangeListener instancesNodeChangeListener = new InstancesNodeChangeListener(this);
        LeaderNodeChangeListener leaderNodeChangeListener = new LeaderNodeChangeListener(this);
        ShardingNodeChangeListener shardingNodeChangeListener = new ShardingNodeChangeListener(this);

        JobTreeCacheListener jobTreeCacheListener = new JobTreeCacheListener(Arrays.asList(configNodeChangeListener,
            instancesNodeChangeListener, leaderNodeChangeListener, shardingNodeChangeListener));
        RegistryCenterConnectionStateListener connectionStateListener =
            new RegistryCenterConnectionStateListener(this);

        TreeCache treeCache = regCenter.getTreeCache("/" + jobName);
        treeCache.getListenable().addListener(jobTreeCacheListener);
        regCenter.getClient().getConnectionStateListenable().addListener(connectionStateListener);

        leaderNode.electLeader();
        instancesNode.persistLocalInstanceOnline();
        started = true;
    }

    public synchronized void pause() {
        log.debug("任务【{}】开始暂停", jobName);
        if (isShutdown() || paused) {
            return;
        }
        paused = true;
        shardingJobExecutor.finishAllShardingItems();
        log.debug("任务【{}】暂停成功", jobName);
    }

    public synchronized void resume() {
        log.debug("任务【{}】开始重新启动", jobName);
        if (isShutdown() || !isPaused()) {
            log.warn("任务【{}】已关闭或未暂停", jobName);
            return;
        }
        paused = false;
        List<String> shardingItems = getInstanceShardingItems(ShardingJobRegistry.getInstance().getLocalInstanceId());
        if (shardingItems.isEmpty()) {
            log.warn("获取不到任务【{}】分片", jobName);
        }
        for (String shardingItem : shardingItems) {
            shardingJobExecutor.execute(new WaitingExecuteShardingItem(shardingItem, shardingNode));
        }
        log.debug("任务【{}】重新启动成功", jobName);
    }

    public synchronized void shutdown() {
        if (!started) {
            return;
        }
        started = false;
        ShardingJobRegistry.getInstance().shutdown(jobName);
        shardingJobExecutor.shutdown();
        instancesNode.removeLocalInstance();
        leaderNode.tryAbdication();
    }

    public boolean isPaused() {
        return paused;
    }

    public boolean isShutdown() {
        return !started;
    }

    public void addShardingItem(String shardingItem, String instanceId) {
        if (isPaused() || isShutdown()) {
            return;
        }
        if (!Objects.equals(instanceId, ShardingJobRegistry.getInstance().getLocalInstanceId())) {
            return;
        }
        shardingJobExecutor.execute(new WaitingExecuteShardingItem(shardingItem, shardingNode));
    }

    public void removeShardingItem(String shardingItem, String instanceId) {
        if (isPaused() || isShutdown()) {
            return;
        }
        if (!Objects.equals(instanceId, ShardingJobRegistry.getInstance().getLocalInstanceId())) {
            return;
        }
        shardingJobExecutor.finishShardingItem(shardingItem);
    }

    public void clearRunningInfo() {
        if (isShutdown()) {
            return;
        }
        List<String> localShardingItems = getInstanceShardingItems(ShardingJobRegistry.getInstance().getLocalInstanceId());
        shardingNode.clearShardingItemRunning(localShardingItems);
    }

    public void shardingFailover(String crashInstanceId) {
        log.debug("任务【{}】开始故障转移", jobName);
        if (isShutdown() || isPaused()) {
            log.warn("任务【{}】已经关闭，无需处理故障转移", jobName);
            return;
        }
        if (!leaderNode.isLeaderUntilBlock()) {
            log.warn("当前实例【{}】不是主节点，无需处理任务【{}】的实例【{}】的故障转移",
                ShardingJobRegistry.getInstance().getLocalInstanceId(), jobName, crashInstanceId);
            return;
        }
        List<String> failoverItems = getInstanceShardingItems(crashInstanceId);
        if (failoverItems.isEmpty()) {
            log.warn("实例【{}】没有负责的分片", jobName);
            return;
        }
        List<String> instanceIds = instancesNode.getAllInstanceIds();
        if (instanceIds.isEmpty()) {
            log.warn("任务【{}】没有可用的实例", jobName);
            return;
        }
        while (!failoverItems.isEmpty()) {
            for (String instanceId : instanceIds) {
                if (failoverItems.isEmpty()) {
                    break;
                }
                String failOverItem = failoverItems.remove(0);
                shardingNode.changeShardingInstance(failOverItem, instanceId);
            }
        }
        log.debug("任务【{}】结束故障转移", jobName);
    }

    public void reSharding() throws RegException {
        log.debug("任务【{}】开始重新分片", jobName);
        if (isShutdown()) {
            log.warn("任务【{}】已经关闭，无需处理重新分片", jobName);
            return;
        }
        if (!leaderNode.isLeaderUntilBlock()) {
            log.warn("当前实例【{}】不是主节点，无需处理任务【{}】重新分片", ShardingJobRegistry.getInstance().getLocalInstanceId(), jobName);
            return;
        }
        // 设置分片执行中标识
        setReShardingFlag();
        // 暂停自己的任务
        // FIXME 为啥创建临时节点，其他实例能收到事件，自己的实例收不到呢？？？
        // 是因为当前正在处理事件回调，回调没处理完，所以不会处理下一个回调，所以自己的任务不会被暂停，只能手动调用pause？
        // 要考虑事件回调的处理改成异步的？
        pause();
        // 加载最新的配置
        config = configNode.load(false);
        // 等待执行中的分片任务停下来
        shardingNode.waitingOtherShardingItemCompleted(config);
        // 获取全部实例id
        List<String> instanceIds = instancesNode.getAllInstanceIds();
        if (instanceIds.isEmpty()) {
            log.warn("任务【{}】没有可用的实例", jobName);
            return;
        }
        // 重置分片信息
        shardingNode.resetShardingInfo(config.getShardingItems());
        // 重新分配分片
        Map<String, List<String>> shardingResults = shardingNode.sharding(instanceIds, config.getShardingItems());
        // 创建分片实例节点
        createShardingItem(shardingResults);
        // 删除分片标识
        shardingNode.clearShardingProcessing();
        log.debug("任务【{}】结束重新分片", jobName);
    }

    private void setReShardingFlag() {
        if (!leaderNode.isLeaderUntilBlock()) {
            return;
        }
        shardingNode.setShardingProcessing();
    }

    private List<String> getInstanceShardingItems(String instanceId) {
        if (isShutdown()) {
            return Collections.emptyList();
        }
        List<String> result = new LinkedList<>();

        List<String> shardingItems = configNode.load(true).getShardingItems();
        for (String shardingItem : shardingItems) {
            if (instanceId.equals(shardingNode.getShardingItemInstanceId(shardingItem))) {
                result.add(shardingItem);
            }
        }
        return result;
    }

    private void createShardingItem(Map<String, List<String>> shardingResults) throws RegException {
        try {
            CuratorTransactionFinal curatorTransactionFinal = regCenter.getClient().inTransaction().check().forPath("/").and();
            for (Map.Entry<String, List<String>> entry : shardingResults.entrySet()) {
                for (String shardingItem : entry.getValue()) {
                    curatorTransactionFinal = curatorTransactionFinal
                        .create()
                        .forPath(AbstractNode.getFullPath(jobName, shardingNode.getShardingItemInstancePath(shardingItem)), entry.getKey().getBytes())
                        .and();
                }
            }
            curatorTransactionFinal.commit();
        } catch (Exception e) {
            RegExceptionHandler.handleException(e);
        }
    }
}
